package com.mocha.client;

import com.mocha.common.MqCommon;
import com.rabbitmq.client.*;
import com.rabbitmq.client.impl.DefaultExceptionHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeoutException;

public class RabbitMQClient {

    private static final Logger LOGGER = LoggerFactory.getLogger(RabbitMQClient.class);

    private String virtualHost;
    private String userName;
    private String password;
    //支持多个，逗号分隔
    private String address;//ip:port,ip:port
    //自动恢复开关
    private boolean automaticRecoveryEnabled = true;
    //定义网络事件间隔长度
    private int networkRecoveryInterval = 3000;
    //拓扑恢复开关
    private boolean topologyRecoveryEnabled = true;

    //amqp-client 异常处理
    private ExceptionHandler exceptionHandler;

    private ExecutorService executorService;

    private ConnectionFactory connectionFactory;
    private Connection connection;

    private boolean persistence = false;

    private static final ThreadLocal<Channel> CHANNEL_THREAD_LOCAL = new ThreadLocal<>();
    private static final String DELAY_EXCHANGE_NAME = "delay" + MqCommon.AT + "gvxcgewg";


    /**
     * init ConnectionFactory 和 Connection
     * @throws IOException
     * @throws TimeoutException
     */
    public void init() throws IOException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setVirtualHost(virtualHost);
        connectionFactory.setUsername(userName);
        connectionFactory.setPassword(password);

        connectionFactory.setAutomaticRecoveryEnabled(automaticRecoveryEnabled);
        connectionFactory.setNetworkRecoveryInterval(networkRecoveryInterval);
        connectionFactory.setTopologyRecoveryEnabled(topologyRecoveryEnabled);

        if (exceptionHandler == null) {
            exceptionHandler = new DefaultExceptionHandler();
        }
        connectionFactory.setExceptionHandler(exceptionHandler);
        connection = connectionFactory.newConnection(executorService, Address.parseAddresses(address));
    }

    /**
     * 获取Channel
     * @return channel
     * @throws IOException
     */
    public Channel getChannel() throws IOException {
        Channel channel = CHANNEL_THREAD_LOCAL.get();
        if (channel == null || !channel.isOpen()) {
            channel = connection.createChannel();
            CHANNEL_THREAD_LOCAL.set(channel);
        }
        return channel;
    }

    public void closeChannel() throws Exception {
        Channel channel = CHANNEL_THREAD_LOCAL.get();
        CHANNEL_THREAD_LOCAL.remove();
        if (channel != null) {
            try {
                channel.close();
            } catch (Exception  e) {
                LOGGER.error("close channel failed , {}", e.getLocalizedMessage());
                throw e;
            }
        }
    }

    public void closeConnection() throws IOException {
        if (connection != null) {
            try {
                connection.close();
            } catch (Exception e) {
                LOGGER.error("close connection failed , {}", e.getLocalizedMessage());
                throw e;
            }
        }
    }

    public <T> T execute(RabbitMQCallback<T> rabbitMQCallback) throws IOException {
        Channel channel = getChannel();
        return rabbitMQCallback.callback(channel);
    }

    /**
     *
     * @param queue 队列名称
     * @param message 消息名称
     * @param persistence 持否持久化
     * @throws IOException
     */
    private void push(String queue, byte[] message, boolean persistence) throws IOException {
        Channel channel = getChannel();
        int deliveryMode = persistence ? 2 : 1;
        AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
        AMQP.BasicProperties properties = builder.deliveryMode(deliveryMode).build();
        //不通过代码来创建queue
        /*if (queueDeclare) {
            channel.queueDeclare(queue, persistence, false, false, null);
        }*/
        channel.basicPublish("", queue, properties, message);
    }

    /**
     * 指定队列发送消息，不持久化，不声明queue
     * @param queue
     * @param message
     * @throws IOException
     */
    public void push(String queue,String message) throws IOException {
        push(queue, message.getBytes(MqCommon.UTF8), false);
    }

    /**
     * 指定队列发送消息，持久化，不声明queue
     * @param queue
     * @param message
     * @throws IOException
     */
    public void pushWithPersistance(String queue, String message) throws IOException {
        push(queue, message.getBytes(MqCommon.UTF8), true);
    }


    /**
     * 通过 TTL & DLE 实现延迟队列
     * @param queue
     * @param message
     * @param delayMillisecond
     * @param persistence
     * @throws Exception
     */
    private void pushDelayWithTTL_DLE(String queue, byte[] message, Long delayMillisecond, boolean persistence) throws Exception {
        if (delayMillisecond == null || delayMillisecond < 0) {
            throw new Exception("delayMillisecond must > 0");
        }
        int deliverMode = persistence ? 2 : 1;
        Channel channel = getChannel();
        channel.queueDeclare(queue, true, false, false, null);
        HashMap<String, Object> arguments = new HashMap<>();
        arguments.put("x-dead-letter-exchange", "");
        arguments.put("x-dead-letter-routing-key", queue);
        String tmpQueue = queue + MqCommon.AT + "tmp";
        channel.queueDeclare(tmpQueue, true, false, false, arguments);

        AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
        AMQP.BasicProperties properties = builder.expiration(delayMillisecond.toString()).deliveryMode(deliverMode).build();
        channel.basicPublish("", tmpQueue, properties, message);

    }

    /**
     * 通过插件实现延迟队列
     * @param queue
     * @param message
     * @param delayMillisecond
     * @param persistence
     * @throws IOException
     */
    private void pushDelayWithPlugin(String queue, byte[] message, long delayMillisecond, boolean persistence) throws IOException {
        int deliveryMode = persistence ? 2 : 1;
        Channel channel = getChannel();

        Map<String, Object> exchangeArguments = new HashMap<>();
        exchangeArguments.put("x-delay-type", "direct");
        channel.exchangeDeclare(DELAY_EXCHANGE_NAME, "x-delayed-message", false, false, exchangeArguments);
        channel.queueDeclare(queue, true, false, false, null);
        channel.queueBind(queue, DELAY_EXCHANGE_NAME, queue);

        Map<String, Object> headers = new HashMap<>();
        headers.put("delay", delayMillisecond);
        AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder().headers(headers).deliveryMode(deliveryMode);
        AMQP.BasicProperties properties = builder.build();

        channel.basicPublish(DELAY_EXCHANGE_NAME, queue, properties, message);

    }

    public void pushDelayWithMillisecond(String queue, String message, long delayMillisecond,boolean withPlugin,boolean persistence) throws Exception {
        if (delayMillisecond <= 0) {
            throw new Exception("delayMillisecond must be after the current system time");
        }
        if (withPlugin) {
            pushDelayWithPlugin(queue,message.getBytes(MqCommon.UTF8),delayMillisecond,persistence);
        }else {
            pushDelayWithTTL_DLE(queue, message.getBytes(MqCommon.UTF8), delayMillisecond, persistence);
        }
    }

    public void pushDelayWithDate(String queue, String message, Date date,boolean withPlugin,boolean persistence) throws Exception {
        long current = System.currentTimeMillis();
        long delayMillisecond = date.getTime() - current;
        if (delayMillisecond <= 0) {
            throw new Exception("delayMillisecond must be after the current system time");
        }
        pushDelayWithMillisecond(queue, message, delayMillisecond, withPlugin, persistence);
    }

    public byte[] pop(String queue) throws IOException {
        Channel channel = getChannel();
        channel.queueDeclare(queue, false, false, false, null);
        GetResponse response = channel.basicGet(queue, true);
        if (response == null) {
            return null;
        }
        return response.getBody();
    }

    public void popWithConsumer(String queue, int perfetchCount, Consumer consumer) throws IOException {
        Channel channel = getChannel();
        channel.queueDeclare(queue, true, false, false, null);
        channel.basicQos(perfetchCount);
        channel.basicConsume(queue, consumer);
    }


    public void popDelayWithConsumer(String queue, boolean autoAck, int perfetchCount, Consumer consumer) throws IOException {
        Channel channel = getChannel();
        channel.queueDeclare(queue, true, false, false, null);
        HashMap<String, Object> arguments = new HashMap<>();
        arguments.put("x-dead-letter-exchange", "");
        arguments.put("x-dead-letter-routing-key", queue);
        String tmpQueue = queue + MqCommon.AT + "tmp";
        channel.queueDeclare(tmpQueue, true, false, false, arguments);
        channel.basicQos(perfetchCount);
        channel.basicConsume(queue, autoAck, consumer);
    }

    public void popDelayWithConsumerByPlugin(String queueName, boolean autoAck, int prefetchCount, Consumer consumer) throws Exception {
        Channel channel = getChannel();
        // 声明exchange
        Map<String, Object> arguments = new HashMap<String, Object>(1);
        arguments.put("x-delayed-type", "direct");
        channel.exchangeDeclare(DELAY_EXCHANGE_NAME, "x-delayed-message", true, false, arguments);
        // 声明队列
        channel.queueDeclare(queueName, true, false, false, null);
        // 绑定队列
        channel.queueBind(queueName, DELAY_EXCHANGE_NAME, queueName);
        channel.basicQos(prefetchCount);
        channel.basicConsume(queueName, autoAck, consumer);
    }

    public void publish(String exchangeName, byte[] message) throws IOException {
        Channel channel = getChannel();
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT.getType(), true);
        channel.basicPublish(exchangeName, "", null, message);
    }

    public void pulishStr(String exchangeName, String message) throws IOException {
        publish(exchangeName, message.getBytes(MqCommon.UTF8));
    }

    public void subscribeWithConsumer(String exchangeName, int perfetchCount, Consumer consumer) throws IOException {
        Channel channel = getChannel();
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT, true);
        String queue = channel.queueDeclare().getQueue();
        channel.queueBind(queue, exchangeName, "");
        channel.basicQos(perfetchCount);
        channel.basicConsume(queue, true, consumer);
    }

    public String getVirtualHost() {
        return virtualHost;
    }

    public void setVirtualHost(String virtualHost) {
        this.virtualHost = virtualHost;
    }

    public String getUserName() {
        return userName;
    }

    public void setUserName(String userName) {
        this.userName = userName;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public String getAddress() {
        return address;
    }

    public void setAddress(String address) {
        this.address = address;
    }

    public boolean isAutomaticRecoveryEnabled() {
        return automaticRecoveryEnabled;
    }

    public void setAutomaticRecoveryEnabled(boolean automaticRecoveryEnabled) {
        this.automaticRecoveryEnabled = automaticRecoveryEnabled;
    }

    public int getNetworkRecoveryInterval() {
        return networkRecoveryInterval;
    }

    public void setNetworkRecoveryInterval(int networkRecoveryInterval) {
        this.networkRecoveryInterval = networkRecoveryInterval;
    }

    public boolean isTopologyRecoveryEnabled() {
        return topologyRecoveryEnabled;
    }

    public void setTopologyRecoveryEnabled(boolean topologyRecoveryEnabled) {
        this.topologyRecoveryEnabled = topologyRecoveryEnabled;
    }

    public ExceptionHandler getExceptionHandler() {
        return exceptionHandler;
    }

    public void setExceptionHandler(ExceptionHandler exceptionHandler) {
        this.exceptionHandler = exceptionHandler;
    }

    public ExecutorService getExecutorService() {
        return executorService;
    }

    public void setExecutorService(ExecutorService executorService) {
        this.executorService = executorService;
    }

    public ConnectionFactory getConnectionFactory() {
        return connectionFactory;
    }

    public void setConnectionFactory(ConnectionFactory connectionFactory) {
        this.connectionFactory = connectionFactory;
    }

    public Connection getConnection() {
        return connection;
    }

    public void setConnection(Connection connection) {
        this.connection = connection;
    }

    public boolean isPersistence() {
        return persistence;
    }

    public void setPersistence(boolean persistence) {
        this.persistence = persistence;
    }















}
